package com.zlm.app;

import com.zlm.bean.PageViewCount;
import com.zlm.bean.UserBehavior;
import com.zlm.util.MyRedisUtils;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import scala.tools.nsc.doc.html.Page;

import java.io.Serializable;
import java.net.URL;
import java.util.Random;

/**
 * Author: Harbour
 * Date: 2021-05-16 19:49
 * Desc:
 */
public class UniqueVisitorApp {



    public static void main(String[] args) throws Exception {
        // step 1 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        // step 2 获取数据源
        URL resource = UniqueVisitorApp.class.getResource("/UserBehavior.csv");
        DataStream<String> inputDataStream = env.readTextFile(resource.getPath());

        // step 3 转换数据类型
        DataStream<UserBehavior> userBehaviorDataStream = inputDataStream
                .filter(data -> "pv".equalsIgnoreCase(data.split(",")[3]))
                .map(new MapFunction<String, UserBehavior>() {
                    @Override
                    public UserBehavior map(String value) throws Exception {
                        String[] fields = value.split(",");
                        return new UserBehavior(
                                new Long(fields[0]),
                                new Long(fields[1]),
                                new Integer(fields[2]),
                                fields[3],
                                new Long(fields[4])
                        );
                    }
                })
                .assignTimestampsAndWatermarks(new AscendingTimestampExtractor<UserBehavior>() {
                    @Override
                    public long extractAscendingTimestamp(UserBehavior element) {
                        return element.getTimestamp() * 1000L;
                    }
                });

        userBehaviorDataStream.timeWindowAll(Time.hours(1))
                .trigger(new MyTrigger())
                .process(new MyUniqueVisitorWithBloomProcessFunctions())
                .print();

        env.execute();
    }

    private static class MyUniqueVisitorWithBloomProcessFunctions extends ProcessAllWindowFunction<UserBehavior, PageViewCount, TimeWindow> {

        private Jedis jedis;
        private final MyBloomFilter bloomFilter = new MyBloomFilter(1 << 29);
        private static final String uvCountMapName = "uvCount";

        @Override
        public void open(Configuration parameters) throws Exception {
            JedisPool jedisClient = MyRedisUtils.getJedisClient();
            jedis = jedisClient.getResource();
        }

        @Override
        public void process(Context context, Iterable<UserBehavior> elements, Collector<PageViewCount> out) throws Exception {
            long windowEnd = context.window().getEnd();
            String bitmapKey = String.valueOf(windowEnd);
            String uvCountKey = String.valueOf(windowEnd);

            String userId = elements.iterator().next().getUserId().toString();
            Long offset = bloomFilter.hashCode(userId, 61);
            // 用redis的get bit方法判断是否存在
            Boolean isExist = jedis.getbit(bitmapKey, offset);

            if (!isExist) {
                jedis.setbit(bitmapKey, offset, true);
                // uvCount 初始值
                long uvCount = 0L;
                String uvCountStr = jedis.hget(uvCountMapName, uvCountKey);
                if (uvCountStr != null && !"".equals(uvCountStr)) {
                    uvCount = Long.parseLong(uvCountStr);
                }
                jedis.hset(uvCountMapName, uvCountKey, String.valueOf(uvCount + 1));
                out.collect(new PageViewCount(uvCountKey, windowEnd, uvCount + 1L));
            }
        }


        @Override
        public void close() throws Exception {
            jedis.close();
        }
    }

    /**
     * 自定义布隆过滤器
     */
    private static class MyBloomFilter implements Serializable {
        private final Integer cap;

        public MyBloomFilter(Integer cap) {
            this.cap = cap;
        }

        public Long hashCode(String value, Integer seed) {
            long result = 0L;
            for (int i = 0; i < value.length(); i++) {
                result = result * seed + value.charAt(i);
            }
            return result & (cap - 1);
        }
    }

    /**
     * 定义一个裁剪类  trigger() —— 触发器 定义 window 什么时候关闭， 触发计算并输出结果
     */
    private static class MyTrigger extends Trigger<UserBehavior, TimeWindow> {

        @Override
        public TriggerResult onElement(UserBehavior element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
            return TriggerResult.FIRE_AND_PURGE; // evaluates the window function and emits the window
        }

        @Override
        public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
            return TriggerResult.CONTINUE;  //No action is taken on the window
        }

        @Override
        public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
            return TriggerResult.CONTINUE;
        }

        @Override
        public void clear(TimeWindow window, TriggerContext ctx) throws Exception {

        }
    }
}
